package com.arica.mq.consumer.listener;

import com.arica.mq.consumer.processor.ProcessorFactory;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.util.CollectionUtils;

import javax.annotation.PreDestroy;

//@Component
//@DependsOn(value = "mQConsumerProperties")
public class BaseConsumerBean implements ApplicationListener<ContextRefreshedEvent> {
    private static Logger logger = LoggerFactory.getLogger(BaseConsumerBean.class);

    private static final String MQ_ON_OFF_CONFIG = "mqConsumerOnOffConfig";

    private static final String TRUE = "true";

    private static final String ENV_ONLINE = "online";

    private static final String ENV_GRAY = "gray";

    private DefaultMQPushConsumer consumer;

//    @Autowired
    private BaseMQConsumerProperties commonProperties;

    @Autowired
    private BaseMessageListener baseMessageListener;

    @Autowired
    private ProcessorFactory processorFactory;

    public void init() {
    }

    @PreDestroy
    public void destroy() {
        if (consumer != null) {
            consumer.shutdown();
        }
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        if (event.getApplicationContext().getParent() != null) {
            return;
        }


        //获取系统MQ监听是否开启的系统配置项，如果开启，则监听mq
//        String mqOnOffConfig = System.getProperty(MQ_ON_OFF_CONFIG);

        String env = commonProperties.getEnv();

        if (StringUtils.isBlank(env)) {
            throw new RuntimeException("启动 consumer 失败，原因: env 为null");
        }

//        logger.info("MQ_ON_OFF_CONFIG:{},env:{}",mqOnOffConfig,env);
//
//        if (!ENV_ONLINE.equals(env) && !ENV_GRAY.equals(env)){
//            if (!TRUE.equals(mqOnOffConfig)) {
//                logger.warn("未设置 MQ_ON_OFF_CONFIG 且非online 环境，不启动 consumer");
//                return;
//            }
//        }

//        // 关闭，则不启动
        if (!commonProperties.isOnOff()) {
            logger.warn("commonProperties 设置为 off，不启动 consumer");
            return;
        }
//
//        // 判断 Properties 是否存在
//        if (commonProperties.getProperties() == null) {
//            logger.error("启动 consumer 失败，原因: Properties 为null");
//            throw new RuntimeException("启动 consumer 失败，原因: Properties 为null");
//        }

        // 判断 topicList 是否存在
        if (CollectionUtils.isEmpty(commonProperties.getTopicList())) {
            logger.error("启动 consumer 失败，原因: topicList 为空");
            throw new RuntimeException("启动 consumer 失败，原因: topicList 为空");
        }

        // 将该对象设置到 baseMessageListener 中
        baseMessageListener.setBaseConsumerBean(this);

        // 初始化 consumer，并启动
        try {
            consumer = new DefaultMQPushConsumer(commonProperties.getGroup());
            consumer.setNamesrvAddr(commonProperties.getNameSrvAddr());
            consumer.setConsumeThreadMin(commonProperties.getConsumeThreadNums());
            consumer.setConsumeThreadMax(commonProperties.getConsumeThreadNums());
            consumer.registerMessageListener(baseMessageListener);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            consumer.setMessageModel(commonProperties.getMessageModel());
            // 设置一次消费消息的条数，默认为1条
            consumer.setConsumeMessageBatchMaxSize(1);
                for (String topic : commonProperties.getTopicList()) {
                    String tag = processorFactory.getTagsByTopic(topic);
                    // TODO 这里为啥不允许没有 tag 的订阅？
                    if (StringUtils.isNotBlank(tag)) {
                        consumer.subscribe(topic, tag);
                    }
//                    consumer.subscribe(topic, "*");
                }
                consumer.start();
            logger.info("BaseConsumerBean Consumer Started");
        } catch (Throwable e) {
            logger.error("启动 consumer 失败，Caused by: {}", ExceptionUtils.getStackFrames(e));
//            throw new RuntimeException("启动 consumer 失败", e);
            logger.error("启动 consumer 失败：【{}】", e.getMessage());
        }
    }

    public BaseMQConsumerProperties getCommonProperties() {
        return commonProperties;
    }

    public void setCommonProperties(BaseMQConsumerProperties commonProperties) {
        this.commonProperties = commonProperties;
    }
}
